package net.dreamlu.iot.mqtt.broker.listener;

import lombok.RequiredArgsConstructor;
import net.dreamlu.iot.mqtt.broker.handler.TopicHandler;
import net.dreamlu.iot.mqtt.codec.MqttPublishMessage;
import net.dreamlu.iot.mqtt.codec.MqttQoS;
import net.dreamlu.iot.mqtt.core.server.event.IMqttMessageListener;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.tio.core.ChannelContext;

import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 消息监听
 *
 * @author L.cm
 */
@Service
public class MqttServerMessageListener implements IMqttMessageListener {
    private static final Logger logger = LoggerFactory.getLogger(MqttServerMessageListener.class);

    private final Map<String, TopicHandler> topicHandlerMap = new ConcurrentHashMap<>();

    public MqttServerMessageListener(Collection<TopicHandler> topicHandlers) {
        topicHandlers.stream().forEach(topicHandler -> topicHandlerMap.put(topicHandler.topic(), topicHandler));
    }

    @Override
    public void onMessage(ChannelContext context, String clientId, String topic, MqttQoS qoS, MqttPublishMessage message) {
        logger.info("clientId:{} message:{} payload:{}", clientId, message, new String(message.getPayload(), StandardCharsets.UTF_8));
        if (Objects.isNull(topic)) {
            return;
        }
        TopicHandler topicHandler = topicHandlerMap.get(topic);
        if (topicHandler == null) {
            logger.error("{}:未找到对应的Handler", topic);
            return;
        }
        topicHandler.handle(clientId, message);
    }
}
